package com.atguigu.order.pay.detect;

import com.atguigu.order.pay.detect.bean.OrderEvent;
import com.atguigu.order.pay.detect.bean.ReceiptEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;

public class TransactionPayMatch {
    //定义侧输出流
    private static final OutputTag<OrderEvent> unmatchedPay = new OutputTag<OrderEvent>("unmatched-pay") {
    };
    private static final OutputTag<ReceiptEvent> unmatchedReceive = new OutputTag<ReceiptEvent>("unmatched-Receive") {
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        URL orderResource = OrderPayTimeout.class.getResource("/OrderLog.csv");
        DataStreamSource<String> orderDataStreamSource = env.readTextFile(orderResource.getPath());
        //读取订单事件
        SingleOutputStreamOperator<OrderEvent> orderEventStream = orderDataStreamSource.map(line -> {
            String[] fields = line.split(",");
            return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
            @Override
            public long extractAscendingTimestamp(OrderEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        //仅选取支付事件才有交易ID
        SingleOutputStreamOperator<OrderEvent> paidOrderStream = orderEventStream
                .filter(data -> "pay".equals(data.getEventType()));

        URL receiptResource = OrderPayTimeout.class.getResource("/ReceiptLog.csv");
        DataStreamSource<String> receiptDataStreamSource = env.readTextFile(receiptResource.getPath());
        //读取到账事件
        SingleOutputStreamOperator<ReceiptEvent> receiptEventStream = receiptDataStreamSource.map(line -> {
            String[] fields = line.split(",");
            return new ReceiptEvent(fields[0], fields[1], new Long(fields[2]));
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvent>() {
            @Override
            public long extractAscendingTimestamp(ReceiptEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        //将两条流进行连接合并，进行匹配处理,不匹配的事件输出到侧输出流
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> resultStream = paidOrderStream
                .keyBy(OrderEvent::getTransactionId)
                .connect(receiptEventStream.keyBy(ReceiptEvent::getTransactionId))
                .process(new TxPayMatchDetect());

        resultStream.print("matched pay");
        resultStream.getSideOutput(unmatchedPay).print("unmatched pay");
        resultStream.getSideOutput(unmatchedReceive).print("unmatched receive");

        env.execute();
    }

    //实现自定义CoProcessFunction
    public static class TxPayMatchDetect extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {
        //定义状态,保存当前已经到来的订单支付事件和到账事件
        ValueState<OrderEvent> payState;
        ValueState<ReceiptEvent> receiptState;

        @Override
        public void open(Configuration parameters) throws Exception {
            payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("pay", OrderEvent.class));
            receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receipt", ReceiptEvent.class));
        }

        @Override
        public void processElement1(OrderEvent pay, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //订单支付事件来了,是否已经有对应的到账事件
            ReceiptEvent receipt = receiptState.value();
            if (receipt != null) {
                out.collect(new Tuple2<>(pay, receipt));
                payState.clear();
                receiptState.clear();
            } else {
                //注册一个定时器开始等待，等待5秒钟
                ctx.timerService().registerEventTimeTimer((pay.getTimestamp() + 5) * 1000L);
                //更新状态
                payState.update(pay);
            }

        }

        @Override
        public void processElement2(ReceiptEvent receipt, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //到账事件
            OrderEvent pay = payState.value();
            if (pay != null) {
                //如果pay不为空，说明支付事件已经来过了
                out.collect(new Tuple2<>(pay, receipt));
                payState.clear();
                receiptState.clear();
            } else {
                //如果pay没来，注册一个定时器开始等待，等待3秒钟
                ctx.timerService().registerEventTimeTimer((receipt.getTimestamp() + 3) * 1000L);
                //更新状态
                receiptState.update(receipt);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //定时器触发，有可能有一个事件没来，也有可能是都来过了已经输出并清空状态
            if (payState.value() != null) {
                ctx.output(unmatchedPay, payState.value());
            }
            if (receiptState.value() != null) {
                ctx.output(unmatchedReceive, receiptState.value());
            }

            payState.clear();
            receiptState.clear();
        }
    }
}
